/*-------------------------------------------------------------------------
 *
 * bloomfilter.c
 *
 * Portions Copyright (c) 2018, Tencent OpenTenBase-C Group
 * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * src/include/utils/misc/bloomfilter.h
 *
 *-------------------------------------------------------------------------
 */

#include "utils/bloomfilter.h"
#include "access/parallel.h"

#include <math.h>

/* Calculate the min directory memory usage in log(base 2) */
static int
MinLogSpace(const int64 ndv, const double fpp)
{
	double k = BUCKET_WORDS;
	double m;

	if (ndv <= 0) return 0;

	/* m is the number of bits we would need to get the fpp specified */
	m = -k * ndv / log(1 - pow(fpp, 1.0 / k));

	/* Handle case where ndv == 1 => ceil(log2(m/8)) < 0. */
	return Max(0, (int)(ceil(log2(m / 8))));
}

/* Get 32 more bits of randomness from a 32-bit hash */
static inline uint32
Rehash32to32(const uint32 hash)
{
  /* Constants generated by uuidgen(1) with the -r flag */
  static const uint64 m = 0x7850f11ec6d14889ull, a = 0x6773610597ca4c63ull;
  /*
   * This is strongly universal hashing following Dietzfelbinger's "Universal hashing
   * and k-wise independent random variables via integer arithmetic without primes". As
   * such, for any two distinct uint32_t's hash1 and hash2, the probability (over the
   * randomness of the constants) that any subset of bit positions of
   * Rehash32to32(hash1) is equal to the same subset of bit positions
   * Rehash32to32(hash2) is minimal.
   */
  return ((uint64)(hash) * m + a) >> 32;
}

BlockBloomFilter
BlockBloomFilterInit(double nrows, double fpp)
{
	BlockBloomFilter filter = NULL;
	int logMemorySize;
	int logNumBuckets;
	int64 allocSize;

	/* Calculate the min memory space in log(base 2) */
	logMemorySize = MinLogSpace((int64)nrows, fpp);

	/* Since log_bufferpool_space is in bytes, we need to convert it to the number of tiny
	 * Bloom filters we will use. */
	logNumBuckets = Max(1, logMemorySize - LOG_BUCKET_BYTE_SIZE);

	if (logNumBuckets > 20)
	{
		elog(WARNING, "Bloom filter: log_num_buckets exceed 20, give up using bloom filter.");
		return NULL;
	}
	allocSize = 1uLL << (logNumBuckets + LOG_BUCKET_BYTE_SIZE);
	allocSize += offsetof(BlockBloomFilterData, directory);

	filter = palloc0(allocSize);
	filter->allocSize = allocSize;
	filter->logNumBuckets = logNumBuckets;
	filter->directoryMask = (1ull << Min(63, logNumBuckets)) - 1;

	elog(DEBUG1, "Block bloom filter: logNumBuckets %d, used %ld Bytes", logNumBuckets, allocSize);

	return filter;
}

dsa_pointer
SharedBloomFilterInit(double nrows, double fpp, dsa_area *area)
{
	BlockBloomFilter filter = NULL;
	int logMemorySize;
	int logNumBuckets;
	int64 allocSize;
	dsa_pointer pdsa;

	/* Calculate the min memory space in log(base 2) */
	logMemorySize = MinLogSpace((int64)nrows, fpp);

	/* Since log_bufferpool_space is in bytes, we need to convert it to the number of tiny
	 * Bloom filters we will use. */
	logNumBuckets = Max(1, logMemorySize - LOG_BUCKET_BYTE_SIZE);

	if (logNumBuckets > 20)
	{
		elog(WARNING, "Bloom filter: log_num_buckets exceed 20, give up using bloom filter.");
		return InvalidDsaPointer;
	}
	allocSize = 1uLL << (logNumBuckets + LOG_BUCKET_BYTE_SIZE);
	allocSize += offsetof(BlockBloomFilterData, directory);

	pdsa = dsa_allocate0(area, allocSize);
	filter = (BlockBloomFilter) dsa_get_address(area, pdsa);

	filter->allocSize = allocSize;
	filter->logNumBuckets = logNumBuckets;
	filter->directoryMask = (1ull << Min(63, logNumBuckets)) - 1;

	return pdsa;
}
/*
 * REHASH is used as 8 odd 32-bit unsigned ints.  See Dietzfelbinger et al.'s "A
 * reliable randomized algorithm for the closest-pair problem".
 */
static uint32 REHASH[8] = {0x47b6137bU, 0x44974d91U, 0x8824ad5bU, 0xa2b7289dU,
		0x705495c7U, 0x2df1424bU, 0x9efc4947U, 0x5c6bfb31U};

/*
 * Adds an element to the BloomFilter. The function used to generate 'hash' need not
 * have good uniformity, but it should have low collision probability. For instance, if
 * the set of values is 32-bit ints, the identity function is a valid hash function for
 * this Bloom filter, since the collision probability (the probability that two
 * non-equal values will have the same hash value) is 0.
 *
 * To set 8 bits in an 32-byte Bloom filter, we set one bit in each 32-bit uint32. This
 * is a "split Bloom filter", and it has approximately the same false positive probability
 * as standard a Bloom filter; See Mitzenmacher's "Bloom Filters and Such". It also has
 * the advantage of requiring fewer random bits: log2(32) * 8 = 5 * 8 = 40 random bits for
 * a split Bloom filter, but log2(256) * 8 = 64 random bits for a standard Bloom filter.
 */
void
BlockBloomFilterInsert(BlockBloomFilter filter, const uint32 hash)
{
	uint32 bucketIdx;
	int i;

	bucketIdx = Rehash32to32(hash) & filter->directoryMask;

	/* Iterate each BucketWord to insert hash value */
	for (i = 0; i < BUCKET_WORDS; ++i)
	{
		BucketWord hval =
			(REHASH[i] * hash) >> ((1 << LOG_BUCKET_WORD_BITS) - LOG_BUCKET_WORD_BITS);
		hval = 1U << hval;
		filter->directory[bucketIdx * 8 + i] |= hval;
	}
	filter->ninsert++;
}

/*
 * Finds an element in the BloomFilter, returning true if it is found and false (with
 * high probabilty) if it is not.
 */
bool
BlockBloomFilterFind(BlockBloomFilter filter, const uint32 hash,
					 uint64 *pLookups, uint64 *pMatches)
{
	uint32 bucketIdx;
#ifdef __AVX2__
	static __m256i rehash_vec;
	static __m256i zero ;
	static __m256i one ;
	__m256i hval_vec;
	__m256i directory_vec_avx2;
	__m256i cmp_result;
	static bool loaded = false;

	if (!loaded)
	{
		rehash_vec = _mm256_loadu_si256((__m256i *)REHASH);
		zero = _mm256_setzero_si256();
		one = _mm256_set1_epi32(1);
		loaded = true;
	}
#else
	int i;
#endif

	/* Skip bloom filter calculation if not initialized. */
	if (filter == NULL)
		return true;

	if (pLookups)
		(*pLookups)++;

	bucketIdx = Rehash32to32(hash) & filter->directoryMask;

#ifdef __AVX2__
	/* Iterate each BucketWord to check hash value using AVX2 */
	hval_vec = _mm256_mullo_epi32(rehash_vec, _mm256_set1_epi32(hash));
	hval_vec = _mm256_srli_epi32(hval_vec, (1 << LOG_BUCKET_WORD_BITS) - LOG_BUCKET_WORD_BITS);
	hval_vec = _mm256_sllv_epi32(one, hval_vec);

	// Load the required values from filter->directory using _mm256_loadu_si256
	directory_vec_avx2 = _mm256_loadu_si256((__m256i *)&filter->directory[bucketIdx * BUCKET_WORDS]);

	cmp_result = _mm256_cmpeq_epi32(_mm256_and_si256(hval_vec, directory_vec_avx2), zero);
	if ( _mm256_movemask_epi8(cmp_result) != 0)
		return false;
#else
	/* Iterate each BucketWord to check hash value */
	for (i = 0; i < BUCKET_WORDS; ++i)
	{
		BucketWord hval =
			(REHASH[i] * hash) >> ((1 << LOG_BUCKET_WORD_BITS) - LOG_BUCKET_WORD_BITS);
		hval = 1U << hval;
		if (!(filter->directory[bucketIdx * 8 + i] & hval))
			return false;
	}
#endif

	if (pMatches)
		(*pMatches)++;

	return true;
}

/*
 * Return true if we eliminated most values. The ratio is set to 75% based on
 * our performance test results.
 */
bool
BlockBloomFilterIsEfficient(BlockBloomFilter filter)
{
	bool efficient = true;

	if (filter == NULL)
		return false;

	if (filter->nlookups > 100000 &&
		filter->nlookups > (filter->ninsert * 10))
	{
		efficient = ((double)filter->nmatches / filter->nlookups) < 0.75 ? true : false;
	}

	return efficient;
}

void
MergeParallelBloomFilter(BlockBloomFilter mergeFilter, BlockBloomFilter localFilter)
{
	int index = 0;
	int i, j;

	/* Iterate each BucketWord to insert hash value */
	for (j = 0; j <= localFilter->directoryMask; j++)
	{
		for (i = 0; i < BUCKET_WORDS; ++i)
		{
			mergeFilter->directory[index] |= localFilter->directory[index];
			index++;
		}
	}

	mergeFilter->ninsert += localFilter->ninsert;
}
